package cn.spdb.harrier.server.master.deal;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import cn.spdb.harrier.common.enmus.ExecutionStatus;
import cn.spdb.harrier.common.utils.Host;
import cn.spdb.harrier.common.utils.Stopper;
import cn.spdb.harrier.server.entity.JobExecutionContext;
import cn.spdb.harrier.server.entity.JobStepBean;
import cn.spdb.harrier.server.master.MasterManagerService;

public class DispatcherDeal implements Runnable, Deal {

	private Logger logger = LoggerFactory.getLogger("Master-" + this.getClass().getSimpleName());

	private LinkedBlockingQueue<String> signal = new LinkedBlockingQueue<String>();

	private PriorityBlockingQueue<JobExecutionContext> dispatcherQueue = new PriorityBlockingQueue<JobExecutionContext>(
			200, new Comparator<JobExecutionContext>() {
				@Override
				public int compare(JobExecutionContext o1, JobExecutionContext o2) {
					if (o1 == null || o2 == null) {
						return 0;
					}
					if (o1.getPriority() == o2.getPriority()) {
						if (o1.getDispatcherTime() != null && o2.getDispatcherTime() != null) {
							if (o1.getDispatcherTime().equals(o2.getDispatcherTime())) {
								return o1.getJob().compareTo(o2.getJob());
							}
							return o1.getDispatcherTime().compareTo(o2.getDispatcherTime());
						}
					}
					return o2.getPriority() - o1.getPriority();
				}
			});

	private final AtomicBoolean STOP = new AtomicBoolean(false);

	private MasterManagerService masterService;

/**
 * 调度任务执行线程的主方法。
 *
 * <p>该方法负责从调度队列中取出待处理的任务，选择合适的 Worker 主机，
 * 并将任务发送给目标主机执行。它会持续监听信号队列，一旦有新任务到达，
 * 就尝试将其分发出去，直到系统停止或线程被暂停。
 *
 * <h3>主要流程如下：</h3>
 * <ol>
 *   <li>等待信号（signal.take）触发调度</li>
 *   <li>进入调度循环：
 *     <ul>
 *       <li>检查是否被暂停</li>
 *       <li>获取队列顶部的任务</li>
 *       <li>获取并设置任务步骤信息</li>
 *       <li>选择合适的目标主机</li>
 *       <li>更新任务状态为 RUNNING</li>
 *       <li>发送任务上下文到目标主机</li>
 *       <li>根据发送结果更新任务状态和记录</li>
 *     </ul>
 *   </li>
 * </ol>
 */
@Override
public void run() {
    // 主循环：只要系统未停止，就一直运行
    while (Stopper.isRunning()) {
        try {
            // 等待信号到来（用于触发调度）
            signal.take();
        } catch (InterruptedException e1) {
            // 捕获中断异常并打印堆栈信息
            e1.printStackTrace();
        }

        // 同步锁住 dispatcherQueue，确保线程安全操作队列
        synchronized (dispatcherQueue) {
            try {
                // 内部循环，持续处理调度任务
                while (Stopper.isRunning()) {
                    try {
                        // 如果当前线程被暂停（STOP 为 true），则休眠 10 秒后继续等待
                        if (STOP.get()) {
                            Thread.sleep(1000 * 10);
                            continue;
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    // 获取队列头部的任务（不移除）
                    JobExecutionContext jobContext = dispatcherQueue.peek();

                    // 如果任务为空，说明队列已空，跳出循环
                    if (ObjectUtils.isEmpty(jobContext)) {
                        break;
                    }

                    // 获取任务的所有步骤信息
                    List<JobStepBean> stepList = masterService.getJobStepList(
                            jobContext.getPlatform(), jobContext.getSystem(), jobContext.getJob());

                    // 如果任务没有定义任何步骤，标记为失败并插入日志记录
                    if (CollectionUtils.isEmpty(stepList)) {
                        logger.error("job step is null,job:{}", jobContext);
                        dispatcherQueue.remove(jobContext);
                        masterService.conversionFailure(jobContext);
                        masterService.insertJobRecord(jobContext);
                        continue;
                    }

                    // 设置任务步骤列表供后续执行使用
                    jobContext.setStepList(stepList);

                    // 加同步锁防止多个线程同时调度同一个任务
                    synchronized (DispatcherDeal.class) {
                        // 根据负载均衡策略选择一个可用的 Worker 主机
                        Host host = masterService.passDispatcherGetHost(jobContext);

                        // 如果没有找到合适的主机，跳出本次循环等待下一次调度
                        if (ObjectUtils.isEmpty(host)) {
                            logger.debug("job select host is null,job:{}", jobContext);
                            break;
                        }

                        // 增加该主机上的任务计数器（权重管理）
                        masterService.incrementWeightAndJob(host, jobContext);

                        // 设置任务上下文中的主机信息
                        jobContext.setHost(host);
                        jobContext.getUdsJob().setServerName(host.getName());

                        // 生成唯一任务实例 ID
                        long id = masterService.getSingleId();
                        jobContext.setTaskInstanceId(id);

                        // 将任务状态更新为 RUNNING，并写入数据库
                        if (masterService.conversionRuning(jobContext)) {

                            // 发送任务上下文到目标主机
                            if (masterService.sendJobExecutionContext(jobContext)) {
                                // 发送成功：记录日志、清除结束时间、插入任务记录、移除队列任务
                                logger.info("job dispatcher success,job:{}", jobContext);
                                jobContext.getUdsJob().setEndTime(null); // 清除结束时间
                                masterService.insertJobRecord(jobContext); // 插入任务记录
                                dispatcherQueue.remove(jobContext); // 从队列中移除
                            } else {
                                // 发送失败：减少主机任务计数器，更新任务状态为 DISPATCHER
                                logger.error("job dispatcher failure,job:{}", jobContext);
                                masterService.decrementWeightAndJob(jobContext.getHost(), jobContext);
                                masterService.updateJobStatus(ExecutionStatus.DISPATCHER,
                                        jobContext.getPlatform(), jobContext.getSystem(), jobContext.getJob());
                            }
                        } else {
                            // 更新任务状态失败，减少主机任务计数器，重置状态
                            masterService.decrementWeightAndJob(jobContext.getHost(), jobContext);
                            masterService.updateJobStatus(ExecutionStatus.DISPATCHER,
                                    jobContext.getPlatform(), jobContext.getSystem(), jobContext.getJob());
                        }
                    }
                }
            } catch (RuntimeException e) {
                // 捕获运行时异常并打印堆栈信息
                e.printStackTrace();
            }
        }
    }
}


	public void stop() {
		STOP.set(true);
		dispatcherQueue.clear();
	}

	public void start() {
		STOP.set(false);
	}

	public void addQueue(JobExecutionContext jobContext) {
		synchronized (dispatcherQueue) {			
			if (dispatcherQueue.contains(jobContext)) {
				dispatcherQueue.remove();
			}
			dispatcherQueue.offer(jobContext);
		}
		signal.offer(jobContext.getPlatform());
	}

	public DispatcherDeal(MasterManagerService masterService) {
		super();
		this.masterService = masterService;
	}

	public void addSignal(String platfrom) {
		signal.offer(platfrom);
	}
}
